package top.jolyoulu.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

/**
 * @Author: JolyouLu
 * @Date: 2024/5/19 15:07
 * @Description
 */
object Spark01_SparkStreaming_Queue {
  def main(args: Array[String]): Unit = {
    //创建环境对象
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    //参数1：环境配置
    //参数2：批处理的周期，采集的周期
    val ssc = new StreamingContext(sparkConf,Seconds(3))
    //创建一个RDD队列
    val rddQueue: mutable.Queue[RDD[Int]] = new mutable.Queue[RDD[Int]]()
    //创建QueueInputDStream
    val inputStream: InputDStream[Int] = ssc.queueStream(rddQueue, oneAtATime = false)
    //处理队列的数据
    val wordToOne: DStream[(Int, Int)] = inputStream.map((_, 1))
    val wordToCount: DStream[(Int, Int)] = wordToOne.reduceByKey(_ + _)

    wordToCount.print()
    //启动采集器
    ssc.start()

    //执行5次 模拟每隔2秒，往rddQueue中添加数据
    for (i <- 1 to 5){
      rddQueue += ssc.sparkContext.makeRDD(1 to 300,10)
      Thread.sleep(2000)
    }

    //等待采集器的关闭
    ssc.awaitTermination()
  }
}
